package com.hangsu.train.netty.transfer.comsumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.hangsu.train.netty.ClientHolder;
import com.hangsu.train.netty.transfer.Context;
import com.hangsu.train.netty.transfer.ECG;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class ECGConsumer extends Thread {

    private final ObjectMapper objectMapper = new ObjectMapper();

    final Object lock = new Object();

    private boolean suspend=false;

    long delay = TimeUnit.SECONDS.toNanos(1) / 500;

    private static AtomicInteger count = new AtomicInteger(0);

    @Override
    public void run() {
        try {
            long lastTime = System.nanoTime();
            while (true) {
                synchronized (lock){
                    if(suspend) {
                        try {
                            lock.wait();
                        }catch (Exception e){

                        }
                    }
                }
                long currentTime = System.nanoTime();
                if (currentTime - lastTime >= delay) {
                    int c= count.incrementAndGet();
                    if(c>1000){
                        count.set(0);
                    }
                    if(c%100 == 0) {
                        if(Context.queue.size()>1000){
                            delay-=400;
                        }else if(Context.queue.size()<1000) {
                            delay+=400;
                        }
                    }

                    lastTime = currentTime;
                    ECG ecg =    Context.queue.take();
                    ClientHolder.getSocketClient().writeAndFlush(new TextWebSocketFrame(objectMapper.writeValueAsString(ecg))).sync();
                }
            }

        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
    }

    public void  setSuspend(boolean suspend){
        if(!suspend){
            synchronized (lock){
                lock.notify();
            }
        }
        this.suspend=suspend;
    }

}
